package sstable

import (
	"log"
	"lsmtree/config"
	"lsmtree/kv"
	sorttree "lsmtree/sortTree"
	"os"
	"time"
)

func (tree *TableTree) Check() {
	tree.majorCompaction()
}

func (tree *TableTree) majorCompaction() {
	con := config.GetConfig()
	for levelIndex, _ := range tree.levels {
		tableSize := uint32(tree.GetLevelSize(levelIndex) / 1000 / 1000)
		// 当前层的 SSTable 数据达到阈值
		// 或者当前层的 SSTable 占用空间大小到达阈值
		// 则开始执行 compact 操作
		if tree.getCount(levelIndex) > con.PartSize || tableSize > levelMaxSize[levelIndex] {
			tree.majorCompactionLevel(levelIndex)
		}
	}
}

// 压缩当前层的文件到下一层
func (tree *TableTree) majorCompactionLevel(level int) {
	log.Println("Compressing layer ", level, "files")
	start := time.Now()
	defer func() {
		elapse := time.Since(start)
		log.Println("Compressing layer ", level, "files done, elapsed time: ", elapse)
	}()

	log.Printf("Compressing layer %d.db files\r\n", level)

	// 用于加载一个 SSTable 的数据区到缓存中
	tableCache := make([]byte, levelMaxSize[level])
	currentNode := tree.levels[level]

	// 当前层的 SSTable 合并到一个有序二叉树中
	memoryTree := &sorttree.Tree{}
	memoryTree.Init()

	tree.lock.Lock()
	for currentNode != nil {
		table := currentNode.table
		// 将 SSTable 的数据区加载到 tableCache 中
		if int64(len(tableCache)) < table.tableMetaInfo.dataLen {
			tableCache = make([]byte, table.tableMetaInfo.dataLen)
		}

		newSlice := tableCache[0:table.tableMetaInfo.dataLen]
		// 读取 SSTable 的数据区
		if _, err := table.f.Seek(0, 0); err != nil {
			log.Println("Failed to read the data area")
			panic(err)
		}
		if _, err := table.f.Read(newSlice); err != nil {
			log.Println("Failed to read the data area")
			panic(err)
		}

		// 解析 SSTable 的数据区
		for k, position := range table.sparseIndex {
			if !position.Deleted {
				value, err := kv.Decode(newSlice[position.Start : position.Start+position.Len])
				if err != nil {
					log.Fatal(err)
				}
				memoryTree.Set(k, value.Value)
			} else {
				memoryTree.Delete(k)
			}
		}

		currentNode = currentNode.next
	}
	tree.lock.Unlock()

	// 将 SortTree 压缩合并成一个 SSTable
	values := memoryTree.GetValues()
	newLevel := level + 1

	// 最多支持 10 层
	if newLevel > 10 {
		newLevel = 10
	}

	// 创建一个新的 SSTable
	tree.createTable(values, newLevel)
	// 清理该层数据
	oldNode := tree.levels[level]

	if level < 10 {
		tree.levels[level] = nil
		tree.clearLevel(oldNode)
	}

}

func (tree *TableTree) clearLevel(oldNode *TableNode) {
	tree.lock.Lock()
	defer tree.lock.Unlock()

	// 清理当前层的每一个 SSTable
	for oldNode != nil {
		err := oldNode.table.f.Close()
		if err != nil {
			log.Println("Failed to close the SSTable: ", oldNode.table.filePath)
			panic(err)
		}
		err = os.Remove(oldNode.table.filePath)
		if err != nil {
			log.Println("Failed to remove the SSTable: ", oldNode.table.filePath)
			panic(err)
		}

		oldNode.table.f = nil
		oldNode.table = nil
		oldNode = oldNode.next
	}
}
